package com.study.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object SparkStreaming02_Queue {
  def main(args: Array[String]): Unit = {
    //TODO 创建环境对象
    //StreamingContext创建时需要两个参数：环境配置和批量处理的周期（采集周期）
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc =new StreamingContext(sparkConf,Seconds(3))

    //1.创建RDD队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    //2.创建QueueInputDStream
    val inputStream = ssc.queueStream(rddQueue,oneAtATime = false)
    //3.处理队列中的RDD数据
    val mappedStream = inputStream.map((_,1))
    val reducedStream = mappedStream.reduceByKey(_ + _)
    //4.打印结果
    reducedStream.print()
    //5.启动任务
    ssc.start()
    //6.循环创建并向RDD队列中放入RDD
    for (i <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }
    /*
    -------------------------------------------
    Time: 1661068797000 ms
    -------------------------------------------
    (224,1)
    (160,1)
    (296,1)
    (96,1)
    ......
     */
    ssc.awaitTermination()
  }
}
